package com.motu.vertx.module.utility.monitor;

import com.motu.vertx.module.utility.commondata.*;
import com.motu.vertx.module.utility.kafka.KafkaLogManager;
import com.motu.vertx.module.utility.mysql.MysqlClient;
import com.motu.vertx.module.utility.mysql.MysqlTool;
import com.motu.vertx.module.utility.toolset.TimeUtils;
import com.motu.vertx.module.utility.toolset.Tool;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;


public class MonitorManager {

    public final static Logger logger = LoggerFactory.getLogger(MonitorManager.class);

    public static JsonArray protocolBannedList = new JsonArray();// 临时屏蔽协议表
    private static List<BannedUser> bannedList = new ArrayList<>();// 封号列表
    private static List<WhiteList> whiteList = new ArrayList<>();// 白名单列表
    private static List<ProtocolMonitor> protocolMonitorList = new ArrayList<>();// 协议监控
    private static List<ProtocolMonitorHttp> protocolMonitorHttpList = new ArrayList<>(); //http协议监控
    private static ConcurrentHashMap<Integer, Integer> protocolRequestList = new ConcurrentHashMap<>();//协议请求配置表（key=协议码，value=允许并发数量）

    public static void init(List<Future> futures) {
		futures.add(initProtocolBanned());// 初始化临时屏蔽的协议表
		futures.add(initBannedlist());// 初始化封号列表
		futures.add(initWhitelist());// 初始化白名单列表
		futures.add(initProtocolMonitor());// 初始化协议监控表
		futures.add(initProtocolMonitorHttp()); //初始化http协议监控表
		futures.add(initProtocolRequest()); //初始化协议请求配置表
    }

    /**
     * 初始化临时屏蔽的协议表
     */
    public static Future initProtocolBanned() {
		Promise promise = Promise.promise();
        String table = "protocol_banned";
        String sql = "SELECT cmd_code FROM " + table + ";";
        MysqlTool.queryJsonArrayFromDb(MysqlClient.getServerPool(), sql, new JsonArray(), handler -> {
            if (handler.succeeded()) {
                JsonArray tmp = new JsonArray();
                JsonArray result = handler.result();
                logger.info("init " + table + " size:" + result.size());
                for (int i = 0; i < result.size(); i++) {
                    int cmdCode = result.getJsonObject(i).getInteger("cmd_code");
                    logger.info("initProtocolBanned cmdcode:{}", cmdCode);
                    tmp.add(cmdCode);
                }
                protocolBannedList = tmp;
				promise.complete();
            } else {
                logger.error("initProtocolBanned error:" + Tool.getException(handler.cause()));
				promise.fail(handler.cause());
            }
        });
		return promise.future();
    }

    /**
     * 检查协议是否屏蔽
     */
    public static boolean checkProtocolBanned(int cmdCode) {
        return protocolBannedList.contains(cmdCode);
    }

    /***
     * 初始化封号列表
     */
    public static Future initBannedlist() {
        Promise promise = Promise.promise();
        String sql = "select * from banned_user;";
        MysqlTool.queryJsonArrayFromDb(MysqlClient.getServerPool(), sql, new JsonArray(), handler -> {
            if (handler.succeeded()) {
                List<BannedUser> tmp = new ArrayList<>();
                JsonArray list = handler.result();
                logger.info("******** initBannedlist ******** len:" + list.size());
                for (int i = 0; i < list.size(); i++) {
                    tmp.add(new BannedUser(list.getJsonObject(i)));
                }
                bannedList = tmp;
                promise.complete();
            } else {
                logger.error("initBannedlist error:" + Tool.getException(handler.cause()));
                promise.fail(handler.cause());
            }
        });
		return promise.future();
    }

    /***
     * 判断是否在封号列表中
     */
    public static boolean isBanned(String accountId, int platform, long userId, String deviceId, String ip) {
        long curTime = TimeUtils.getCurTime();
        for (int i = 0; i < bannedList.size(); i++) {
            BannedUser buser = bannedList.get(i);
            // 已经过了解封时间
            if (buser.getBanTime() > 0 && buser.getBanTime() < curTime) {
                continue;
            }
            // 封账号
            if (accountId != null && !buser.getAccountId().isEmpty() && accountId.equalsIgnoreCase(buser.getAccountId()) && buser.getType() == 1 && platform == buser.getPlatform()) {
                return true;
            }
            // 封角色
            if (userId > 0 && userId == buser.getUserId()) {
                return true;
            }
            // 封设备
            if (deviceId != null && !buser.getDeviceId().isEmpty() && deviceId.equalsIgnoreCase(buser.getDeviceId()) && buser.getType() == 2) {
                return true;
            }
            // 封ip
            if (ip != null && buser.getType() == 3 && !buser.getIp().isEmpty() && ip.equalsIgnoreCase(buser.getIp())) {
                return true;
            }
        }
        return false;
    }

    /***
     * 初始化白名单
     */
    public static Future initWhitelist() {
        Promise promise = Promise.promise();
        String sql = "select * from white_list;";
        MysqlTool.queryJsonArrayFromDb(MysqlClient.getServerPool(), sql, new JsonArray(), handler -> {
            if (handler.succeeded()) {
                List<WhiteList> tmp = new ArrayList<>();
                JsonArray list = handler.result();
                logger.info("******** initWhitelist ******** len:" + list.size());
                for (int i = 0; i < list.size(); i++) {
                    tmp.add(new WhiteList(list.getJsonObject(i)));
                }
                whiteList = tmp;
                promise.complete();
            } else {
                logger.error("initWhitelist error:" + Tool.getException(handler.cause()));
                promise.fail(handler.cause());
            }
        });
		return promise.future();
    }

    /***
     * 判断是否在白名单中
     */
    public static boolean isWhiteList(String accountId, String ip, int platform) {
        for (int i = 0; i < whiteList.size(); i++) {
            WhiteList wl = whiteList.get(i);
            if (wl.getType() == 0) {//账号白名单
                if (wl.getParams().equalsIgnoreCase(accountId)) {
                    return true;
                }
            } else if (wl.getType() == 1) {//ip白名单
                if (wl.getParams().equalsIgnoreCase(ip)) {
                    return true;
                }
            } else if (wl.getType() == 2) {//platform白名单
                if (wl.getParams().equalsIgnoreCase("" + platform)) {
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * 是否是ip白名单
     */
    public static boolean isIpWhiteList(String ip){
        for(WhiteList wl : whiteList){
            if(wl.getType() == 1){//ip白名单
                if(wl.getParams().equalsIgnoreCase(ip)){
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * 初始化协议监控表
     */
    public static Future initProtocolMonitor() {
		Promise promise = Promise.promise();
        String sql = "select * from protocol_monitor;";
        MysqlTool.queryJsonArrayFromDb(MysqlClient.getServerPool(), sql, new JsonArray(), handler -> {
            if (handler.succeeded()) {
                JsonArray list = handler.result();
                logger.info("******** initProtocolMonitor ******** len:" + list.size());
                List<ProtocolMonitor> temp = new ArrayList<>();
                for (int i = 0, size = list.size(); i < size; i++) {
                    ProtocolMonitor model = new ProtocolMonitor(list.getJsonObject(i));
                    temp.add(model);
                }
                protocolMonitorList = temp;
				promise.complete();
            } else {
                logger.error("initProtocolMonitor error:" + Tool.getException(handler.cause()));
				promise.fail(handler.cause());
            }
        });
		return promise.future();
    }

    /**
     * 是否需要监控协议
     */
    public static boolean isCmdMonitor(int cmdCode, long userId) {
        for (ProtocolMonitor model : protocolMonitorList) {
            if (model.isValid()) {
                if (model.getCmdCode() == 0 && model.getUserId() == 0) {
                    return true;//记录所有人的所有协议
                }

                if (model.getCmdCode() == cmdCode && model.getUserId() == 0) {
                    return true;//记录所有人的某个协议
                }

                if (model.getCmdCode() == 0 && model.getUserId() == userId) {
                    return true;//记录某个人的所有协议
                }

                if (model.getCmdCode() == cmdCode && model.getUserId() == userId) {
                    return true;//记录某个人的某个协议
                }
            }
        }
        return false;
    }

    /**
     * 添加协议监控数据
     */
    public static void addProtocolMonitorData(long userId, int cmdCode, int resCode, byte[] request, byte[] response) {
        if (!MonitorManager.isCmdMonitor(cmdCode, userId)) {
            return;
        }
        long curTime = System.currentTimeMillis();
        JsonArray params = new JsonArray();
        params.add(userId);
        params.add(cmdCode);
        params.add(resCode);
        params.add(request);// 这里是通过kafka写数据库，所以不用Buffer.buffer
        params.add(response);// 这里是通过kafka写数据库，所以不用Buffer.buffer
        params.add(curTime);
        params.add(TimeUtils.getMsTimeFullStr(curTime));

        KafkaLogManager.addProtocolLog(params);
    }

    /**
     * 初始化协议监控表
     */
    public static Future initProtocolMonitorHttp() {
		Promise promise = Promise.promise();
        String sql = "SELECT * FROM protocol_monitor_http;";
        MysqlTool.queryJsonArrayFromDb(MysqlClient.getServerPool(), sql, new JsonArray(), handler -> {
            if (handler.succeeded()) {
                JsonArray list = handler.result();
                logger.info("******** initProtocolMonitorHttp ******** len:" + list.size());
                List<ProtocolMonitorHttp> temp = new ArrayList<>();
                for (int i = 0, size = list.size(); i < size; i++) {
                    ProtocolMonitorHttp model = new ProtocolMonitorHttp(list.getJsonObject(i));
                    if (model.getCmdCode() == 0 && model.getEndTime() == 0) {
                        // 避免出现无限时间监控所有http协议
                        continue;
                    }
                    temp.add(model);
                }
                protocolMonitorHttpList = temp;
				promise.complete();
            } else {
                logger.error("initProtocolMonitor " + handler.cause().getMessage());
				promise.fail(handler.cause());
            }
        });
		return promise.future();
    }

    public static boolean isCmdMonitor(int cmdCode) {
        for (ProtocolMonitorHttp model : protocolMonitorHttpList) {
            if (model.isValid()) {
                if (model.getCmdCode() == 0) {
                    return true;//记录所有人的所有协议
                }
                if (model.getCmdCode() == cmdCode) {
                    return true;//记录所有人的某个协议
                }
            }
        }
        return false;
    }

    public static void addProtocolMonitorDataHttp(int cmdCode, byte[] request, byte[] response) {
		if (!MonitorManager.isCmdMonitor(cmdCode)) {
			return;
		}
        String sql = "INSERT INTO protocol_monitor_data_http(cmd_code,request,response,add_time) VALUES(?,?,?,?);";
        JsonArray params = new JsonArray();
        params.add(cmdCode);
        params.add(request);
        params.add(Buffer.buffer(response));// 如果是直接写数据库，byte[]写入数据库中要转成Buffer，读取的时候用getBuffer，如果是通过kafka写数据库，则不需要转
        params.add(TimeUtils.getCurTime());
        MysqlTool.updateToDb(MysqlClient.getServerPool(), sql, params, res -> {
            if (res.failed()) {
                logger.error("addProtocolMonitorDataHttp " + res.cause().getMessage());
            }
        });
    }

    /**
     * 初始化协议请求配置表
     */
    public static Future initProtocolRequest() {
        String table = "protocol_request";
        Promise promise = Promise.promise();
        String sql = "select * from " + table + ";";
        MysqlTool.queryJsonArrayFromDb(MysqlClient.getServerPool(), sql, new JsonArray(), res -> {
            if (res.succeeded()) {
                try {
                    ConcurrentHashMap<Integer, Integer> tmp = new ConcurrentHashMap<>();
                    JsonArray result = res.result();
                    logger.info("init " + table + " size:" + result.size());
                    for (int i = 0; i < result.size(); i++) {
                        ProtocolRequest entity = new ProtocolRequest(result.getJsonObject(i));
                        tmp.put(entity.getCmdcode(), entity.getNum());
                    }
                    protocolRequestList = tmp;
                    promise.complete();
                } catch (Exception e) {
                    logger.error("init " + table + " error exception:" + Tool.getException(e));
                    promise.fail(e);
                }
            } else {
                logger.error("initProtocolRequest error:" + res.cause().getMessage());
                promise.fail(res.cause());
            }
        });
        return promise.future();
    }

    /**
     * 获取协议允许的并发请求数量
     * @return null表示没配置,0表示不限制
     */
    public static Integer getProtocolRequestNum(int cmdCode) {
        Integer result = protocolRequestList.get(cmdCode);
        return result;
    }

}
